package keter.lab.grpc.cloud.util;

import com.google.common.util.concurrent.*;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;

/**
 * Created by gulixing@msn.com on 2018/3/7.
 */
public class ConcurrentUtil {

    private static final Logger logger = LoggerFactory.getLogger(ConcurrentUtil.class);
    final static int DEFAULT_POOL_SIZE = 2;
    final static String DEFAULT_POOL_NAME = "DEFAULT_WORKER_THREAD";
    /** 线程运行后是否可自动退出：true 退出，false 不退出
     * the JVM shuts down if there are only daemon threads running. */
    final  boolean DEFAULT_DAEMON = true;

    BasicThreadFactory factory;

    private static ConcurrentUtil instance() {
        return new ConcurrentUtil();
    }

    /**
     * 默认线程池
     * @return
     */
    public static ListeningExecutorService service(){
        return ConcurrentUtil.instance().getService();
    }

    /**
     * 默认线程池
     * @param size 线程池工作线程数
     * @return
     */
    public static ListeningExecutorService service(int size){
        return ConcurrentUtil.instance().getService(size);
    }

    public BasicThreadFactory getFactory(){
        return factory;
    }

    protected ListeningExecutorService getService(){
        return getService(DEFAULT_POOL_SIZE);
    }

    protected ListeningExecutorService getService(int poolSize){
        return MoreExecutors.listeningDecorator(getExecutorService(DEFAULT_POOL_NAME+"["+RandomUtils.nextInt()+"]-%d",poolSize,DEFAULT_DAEMON));
    }

    protected ListeningExecutorService getService(String name, int poolSize, boolean exit){
        return MoreExecutors.listeningDecorator(getExecutorService(name,poolSize,exit));
    }

    public static <T> T fetchResult(ListeningExecutorService service, ListenableFuture future) throws InterruptedException, ExecutionException {
        return (T) fetchResults(service,future).get(0);
    }

    public static <T> List<T>  fetchResults(ListeningExecutorService service, ListenableFuture... future) throws InterruptedException, ExecutionException {
        final ListenableFuture allFutures = Futures.allAsList(future);
        final ListenableFuture transform = Futures.transformAsync(allFutures, new AsyncFunction<List<T>, Boolean>() {
            @Override
            public ListenableFuture apply(List<T> results){
                return Futures.immediateFuture(results);
            }
        },service);

        Futures.addCallback(transform, new FutureCallback<List>() {
            @Override
            public void onSuccess(List result) {
                logger.debug("success with:{}",result.get(0));
            }

            @Override
            public void onFailure(Throwable thrown) {
                logger.error("onFailure:{}", thrown.getMessage());
            }
        },service);
        return  (List<T>) transform.get();
    }

    private  ExecutorService getExecutorService(String name, int poolSize, boolean exit){
        factory = getBasicThreadFactory(name,exit);
        return new ThreadPoolExecutor(
                poolSize,
                poolSize,
                0L,TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                factory,
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    private  BasicThreadFactory getBasicThreadFactory(String name, boolean exit){
        return new BasicThreadFactory.Builder()
                .namingPattern(name)
                .daemon(exit)
                .build();
    }
}

